package midas.mw.crt.jxta;

import midas.mw.cfg.CfgError;
import midas.mw.cfg.CfgLogging;
import midas.mw.cfg.log.Logger;
import midas.mw.crt.Constants;
import midas.mw.crt.presence.PresenceKey;
import midas.mw.crt.presence.util.MD5ID;
import net.jxta.document.AdvertisementFactory;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.id.ID;
import net.jxta.peergroup.PeerGroup;
import net.jxta.pipe.InputPipe;
import net.jxta.pipe.PipeMsgEvent;
import net.jxta.pipe.PipeMsgListener;
import net.jxta.pipe.PipeService;
import net.jxta.protocol.PipeAdvertisement;
import java.io.IOException;
import java.util.Date;
import java.util.Iterator;



public class PresencePipe implements PipeMsgListener 
{
	
	private PipeService pipeService;
	private InputPipe inputPipe = null;
	
	private JxtaModule jxta = null;
	/**
	* Constructor for the PipeServer object
	* 
	*/
	public PresencePipe(JxtaModule jxta) 
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "PresencePipe(JxtaModule)");
		
		this.jxta = jxta;
	}
	
	/**
	* Dumps the message content to stdout
	*
	* @param msg the message
	* @param verbose dumps message element content if true
	*/
	/*public static void printMessageStats(Message msg, boolean verbose) 
	{
		try {
			CountingOutputStream cnt;
			ElementIterator it = msg.getMessageElements();
			System.out.println("------------------Begin Message---------------------");
			WireFormatMessage serialed = WireFormatMessageFactory.toWire(
					msg, new MimeMediaType("application/x-jxta-msg"), null);
			System.out.println("Message Size :" + serialed.getByteLength());
			while (it.hasNext()) {
				MessageElement el = it.next();
				String eName = el.getElementName();
				cnt = new CountingOutputStream(new DevNullOutputStream());
				el.sendToStream(cnt);
				long size = cnt.getBytesWritten();
				System.out.println("Element " + eName + " : " + size);
				if (verbose) {
					System.out.println("[" + el + "]");
				}
			}
			System.out.println("-------------------End Message----------------------");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}*/
	/**
	* Creates the input pipe with this as the message listener
	*/
	public InputPipe start(PeerGroup peerGroup) 
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "PresencePipe.start(PeerGroup)");
		
		try {								
			PipeAdvertisement advertisement = (PipeAdvertisement)
			AdvertisementFactory.newAdvertisement(PipeAdvertisement.getAdvertisementType());
			ID presencePipeID = (net.jxta.id.ID)MD5ID.createPipeID(peerGroup.getPeerGroupID(),"PipeService", "Node1");			
			advertisement.setPipeID(presencePipeID);
			advertisement.setType(PipeService.UnicastType);
			advertisement.setName("PresencePipe");
			try {
	        	peerGroup.getDiscoveryService().publish(advertisement);	        	
	        } catch(Exception e){
	            
	            throw new RuntimeException("Critical Error pubishing our own advertisement");
	        }
			// Create the InputPipe and register this for message arrival
			// notification call-back
			pipeService = peerGroup.getPipeService();
			inputPipe = pipeService.createInputPipe(advertisement, this);
		} catch (IOException io) {
			io.printStackTrace();
			return null;
		}
		if (inputPipe == null) {
			CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_FATAL, 
					"Presence.start(). Cannot open InputPipe ", null);			
			return null;
			//System.exit(-1);
		}		
		CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL,
					"Presence.start(). Waiting for msgs on input pipe");
		return inputPipe;
	}
	/**
	* Closes the output pipe and stops the platform
	*/
	public void stop() 
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "PresencePipe.stop()");
		//Close the input pipe
		inputPipe.close();
		//Stop JXTA
	}
	/**
	* PipeMsgListener interface for asynchronous message arrival notification
	*
	* @param event the message event
	*/
	public void pipeMsgEvent(PipeMsgEvent event) 
	{	
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "PresencePipe.pipeMsgEvent(PipeMsgEvent)");
		
		Message msg;
		try {
		// Obtain the message from the event
			msg = event.getMessage();
			if (msg == null) {
				CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_WARNING, 
						"PresencePipe.pipeMsgEvent(). Received an empty message ", null);				
				return;
			}
			// dump the message content to screen
			//printMessageStats(msg, true);
		} catch (Exception e) {
			e.printStackTrace();
			return;
		}
		// get all the message elements
		Message.ElementIterator en = msg.getMessageElements();
		if (!en.hasNext()) {
			return;
		}
		
		String jxtaId 		= null;
		String nodeId 		= null;
		String nodeType		= null;
		String proxyName	= null;
		String ipAddress 	= null;
		MessageElement me1 = msg.getMessageElement(Constants.MIDAS, Constants.MESSAGE_TYPE);
		if (me1.toString().equals(Constants.ACTION_ADD) || 
				me1.toString().equals(Constants.ACTION_DELETE))
		{
			msg.removeMessageElement(me1);
			for (Iterator en2 = msg.getMessageElementsOfNamespace(Constants.MIDAS); en2.hasNext();)
			{
				jxtaId 		= ((MessageElement)en2.next()).toString();
				nodeId 		= ((MessageElement)en2.next()).toString();				
				nodeType 	= ((MessageElement)en2.next()).toString();
				proxyName 	= ((MessageElement)en2.next()).toString();
				ipAddress 	= ((MessageElement)en2.next()).toString();
																
				CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "PresencePipe.pipeMsgEvent(PipeMsgEvent). " + 
						" \n\tType: " + me1.toString() +  
						" \n\tJXTA identifier: " + jxtaId + 
						" \n\tMIDAS node identifier: " + nodeId +
						" \n\tNodeType: " + nodeType +
						" \n\tProxyName: " + proxyName +
						" \n\tIP Addresses: " + ipAddress);				
			} 		
		}
		
		// get the message element in the name space PipeClient.MESSAGE_NAME_SPACE
		//MessageElement msgElement = msg.getMessageElement(MESSAGE_NAME_SPACE);
		// Get message
		if (me1.toString() == null) {
			CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_WARNING, 
					"PresencePipe.pipeMsgEvent(PipeMsgEvent). Null msg received ", null);			
		} 
		else {
			Date date = new Date(System.currentTimeMillis());
			CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgError.SEV_INFORMAL,
					"PresencePipe.pipeMsgEvent(PipeMsgEvent) Message received at :" + date.toString());
			CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgError.SEV_INFORMAL,
					"PresencePipe.pipeMsgEvent(PipeMsgEvent) Message created at :" + me1.toString());
		}
				
		//Se lo mando al resto de los nodos
		String cadenaPresenceKey = jxtaId + "," + nodeId + "," +  nodeType + "," + proxyName + ",0";		
		PresenceKey presenceKey = PresenceKey.create(cadenaPresenceKey);
		if (me1.toString().equals(Constants.ACTION_ADD))
		{
			jxta.getNodePresence().addPeer(presenceKey);
			jxta.getNodePresence().scheduleBulkAnnounce();
		}
		else
			jxta.getNodePresence().removePeer(presenceKey);
	}
}
